package cn.initcap;

import com.google.gson.Gson;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

/**
 * @author initcap
 * @date Created in 2020/7/12 10:23.
 */
public class RabbitMqJob {

    public static void main(String[] args) throws Exception {
        //1.获取flink的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setHost("192.168.31.76")
                .setPort(5672)
                .setUserName("dc3")
                .setPassword("dc3")
                .setVirtualHost("/")
                .build();

        //2.连接socket获取输入的数据(数据源Data Source)
        //2.1. rabbitmq连接的配置，2.rabbitmq的队列名，消费的队列名
        DataStream<String> dataStreamSource = env.addSource(new RMQSource<String>(connectionConfig,
                "dc3.queue.point.value", true, new SimpleStringSchema()));
        dataStreamSource.print();   //输出Source的信息


        //3.数据转换
        //MapFunction:第一个参数是你接收的数据的类型
        //MapFunction:第二个参数是返回数据的类型
        DataStream<PointData> requtapisMessageVo = dataStreamSource.map(new MapFunction<String, PointData>() {
            @Override
            public PointData map(String value) throws Exception {
                Gson gson = new Gson();
                //反序列化,拿到实体(我这里接受的是实体，如果你只是string 直接return就完事了)
                PointData pointData = gson.fromJson(value, PointData.class);
                System.out.println(pointData);
                return pointData;
            }
        });

        //4.sink输出
        // requtapisMessageVo.addSink(new SinkMysql());
        //5.这一行代码一定要实现，否则程序不执行
        env.execute("Socket window coun1t");
    }

}
